package com.o2o.repository;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.NavigableMap;
import java.util.Set;

import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.InclusiveStopFilter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.SingleColumnValueFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Repository;

import com.o2o.base.BaseRepository;
import com.o2o.content.HBaseContent;

@Scope("prototype")
@Repository
public class HBaseRepository extends BaseRepository{
	
	public void selectRowKey(String tablename, String rowKey)throws IOException {
		HTable table = new HTable(configuration, tablename);
		Get g = new Get(rowKey.getBytes());
		Result rs = table.get(g);

		for (KeyValue kv : rs.raw()) {
			System.out.println("--------------------" + new String(kv.getRow())
					+ "----------------------------");
			System.out.println("Column Family: " + new String(kv.getFamily()));
			System.out.println("Column       :" + new String(kv.getQualifier())
					+ "t");
			System.out.println("value        : " + new String(kv.getValue()));
		}
	}

	public void selectRowKeyFamily(String tablename, String rowKey,String family) throws IOException {
		HTable table = new HTable(configuration, tablename);
		Get g = new Get(rowKey.getBytes());
		g.addFamily(Bytes.toBytes(family));
		Result rs = table.get(g);
		for (KeyValue kv : rs.raw()) {
			System.out.println("--------------------" + new String(kv.getRow())
					+ "----------------------------");
			System.out.println("Column Family: " + new String(kv.getFamily()));
			System.out.println("Column       :" + new String(kv.getQualifier())
					+ "t");
			System.out.println("value        : " + new String(kv.getValue()));
		}
	}

	public void selectRowKeyFamilyColumn(String tablename,String rowKey, String family, String column) throws IOException {
		HTable table = new HTable(configuration, tablename);
		Get g = new Get(rowKey.getBytes());
		g.addColumn(family.getBytes(), column.getBytes());

		Result rs = table.get(g);

		for (KeyValue kv : rs.raw()) {
			System.out.println("--------------------" + new String(kv.getRow())
					+ "----------------------------");
			System.out.println("Column Family: " + new String(kv.getFamily()));
			System.out.println("Column       :" + new String(kv.getQualifier())
					+ "t");
			System.out.println("value        : " + new String(kv.getValue()));
		}
	}

	public void selectFilter(String tablename, List<String> arr) throws IOException {
		HTable table = new HTable(configuration, tablename);
		Scan scan = new Scan();// 实例化一个遍历器
		FilterList filterList = new FilterList(); // 过滤器List

		for (String v : arr) { // 下标0为列簇，1为列名，3为条件
			String[] wheres = v.split(",");

			filterList.addFilter(new SingleColumnValueFilter(// 过滤器
					wheres[0].getBytes(), wheres[1].getBytes(),

					CompareOp.EQUAL,// 各个条件之间是" and "的关系
					wheres[2].getBytes()));
		}
		scan.setFilter(filterList);
		ResultScanner ResultScannerFilterList = table.getScanner(scan);
		for (Result rs = ResultScannerFilterList.next(); rs != null; rs = ResultScannerFilterList
				.next()) {
			for (KeyValue kv : rs.list()) {
				System.out.println("--------------------"
						+ new String(kv.getRow())
						+ "----------------------------");
				System.out.println("Column Family: "
						+ new String(kv.getFamily()));
				System.out.println("Column       :"
						+ new String(kv.getQualifier()) + "t");
				System.out.println("value        : "
						+ new String(kv.getValue()));
			}
		}
	}
	/**
	 * 创建一张表，并创建一列簇
	 * @param tableName 表名称
	 * @param columnFamily 列簇名称
	 * @throws IOException
	 */
	public void createHTable(String tableName, String columnFamily) throws IOException {
		try {
			if (admin.tableExists(tableName))
				return;// 判断表是否已经存在
			HTableDescriptor htdesc = this.createHTDesc(tableName);
			htdesc.addFamily(new HColumnDescriptor(columnFamily));
			admin.createTable(htdesc);
		} catch (IOException e) {
			throw e;
		}
	}

	public boolean createHTable(String tableName) throws IOException {
		try {
			if (admin.tableExists(tableName))
				return true;// 判断表是否已经存在
			HTableDescriptor htdesc = this.createHTDesc(tableName);
			admin.createTable(htdesc);
			return true;
		} catch (IOException e) {
			throw e;
		}
	}

//	public void deleteColumn(String tableName, String rowID, String colName,String cluster) throws IOException {
//		try {
//			Delete del = new Delete(rowID.getBytes());
//			if (cluster == null || "".equals(cluster))
//				del.deleteColumn(colName.getBytes());
//			else
//				del.deleteColumn(colName.getBytes(), cluster.getBytes());
//			HTable hTable = this.getHTable(tableName);
//			hTable.delete(del);
//		} catch (IOException e) {
//			throw e;
//		}
//	}
	/**
	 * 根据table表名称，rowKey，以及columnFamily列簇，得到一条对应的key/value形式的map
	 * @param tableName 表名称
	 * @param rowKey rowKey
	 * @param columnFamily 列簇
	 * @return NavigableMap<byte[], byte[]>字段=对应值 
	 * @throws IOException
	 */
	public NavigableMap<byte[], byte[]> getColumnKeyValueMap(String tableName, String rowKey, String columnFamily) throws IOException {
		String[] ids = new String[]{rowKey};
		List<NavigableMap<byte[], byte[]>> columnKeyValueListMap = this.getColumnKeyValueListMap(tableName, ids, columnFamily);
		if(columnKeyValueListMap != null && columnKeyValueListMap.size() != 0){
			return columnKeyValueListMap.get(0);
		}else return null;
	}
	/**
	 * 根据表明,rowkey的数组，以及列簇来获取对应的记录
	 * @param tableName 表名称
	 * @param rowKeys 查询rowKey的string数组
	 * @param columnFamily 列簇
	 * @return List<NavigableMap<byte[], byte[]>>
	 * @throws IOException
	 */
	public List<NavigableMap<byte[], byte[]>> getColumnKeyValueListMap(String tableName, String[] rowKeys, String columnFamily) throws IOException {
		HTable table = new HTable(configuration, tableName);
		List<NavigableMap<byte[], byte[]>> keyValue = Collections.synchronizedList(new ArrayList<NavigableMap<byte[], byte[]>>());
		List<Get> gets = Collections.synchronizedList(new ArrayList<Get>());
		for (String string : rowKeys) {
			Get get = new Get(string.getBytes());
			get.addFamily(Bytes.toBytes(columnFamily));
			gets.add(get);
		}
        long a = System.currentTimeMillis();
        Result[] results = table.get(gets);
        long b = System.currentTimeMillis();
        logger.info("query table timeMillis:"+(b-a)+"ms");
        for (Result result : results) {
        	NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(columnFamily.getBytes());
        	if(familyMap != null)
        		keyValue.add(familyMap);
		}
		return keyValue;
	}
	public Map<String, String> getColumnValue(String tableName, String columnFamily,String qualifier) throws IOException {
		ResultScanner scanner = null;
		try {
			HTable hTable = this.getHTable(tableName);
			scanner = hTable.getScanner(columnFamily.getBytes(), qualifier.getBytes());
			Result rowResult = scanner.next();
			Map<String, String> resultMap = new HashMap<String, String>();
			String row;
			while (rowResult != null) {
				row = new String(rowResult.getRow());
				resultMap.put(row,new String(rowResult.getValue(columnFamily.getBytes(),qualifier.getBytes())));
				rowResult = scanner.next();
			}
			return resultMap;
		} catch (IOException e) {
			throw e;
		} finally {
			if (scanner != null) {
				scanner.close();// 一定要关闭
			}
		}
	}
	/**
	 * 根据table，rowKey，columnFamily以及qualifier来获取value值
	 * @param tableName 表名称
	 * @param rowKey 行key
	 * @param columnFamily 列簇
	 * @param qualifier dataKey
	 * @return String value值，若没有，则返回""
	 * @throws IOException
	 */
	public String getValue(String tableName, String rowKey, String columnFamily,String qualifier) throws IOException {
		try {
			HTable hTable = this.getHTable(tableName);
			Get get = new Get(rowKey.getBytes());
			Result result = hTable.get(get);
			byte[] b = result.getValue( columnFamily.getBytes(),qualifier.getBytes());
			if (b == null)
				return "";
			else
				return new String(b);
		} catch (IOException e) {
			throw e;
		}
	}
	
	/**
	 * 插入或更新某一行rowKey，多个column记录
	 * @param tableName table名称
	 * @param rowKey 记录主键id，mysql持久化后得到的id
	 * @param columnFamily 列簇
	 * @param column  一个map，key为要存储的qualifier，value为要存储的value
	 * @throws IOException
	 */
	public void insertAndUpdate(String tableName, String rowKey, String columnFamily,Map<String,String> column) throws IOException {
		HTable table = this.getHTable(tableName);
		List<Put> puts = new ArrayList<Put>();
		Set<Entry<String, String>> entrySet = column.entrySet();
		logger.info("save rowKey:"+rowKey);
		for (Entry<String, String> entry : entrySet) {
			Put p = new Put(Bytes.toBytes(rowKey));
			p.add(Bytes.toBytes(columnFamily), Bytes.toBytes(entry.getKey()),Bytes.toBytes(entry.getValue()));
			puts.add(p);
		}
		table.put(puts);
	}
	/**
	 * 根据给出的tableName,列簇,开始的rowKey，结束rowKey，偏移量offset以及限制数量limit获取
	 * 符合条件的columns
	 * @param tableName 表名称
	 * @param columnFamily 列簇
	 * @param startKey 开始的rowKey
	 * @param stopKey 结束的rowKey
	 * @param offset 偏移量
	 * @param limit 限制数量，注意，如果有足够的limit，则会返回limit+1条数据
	 * @param rowFlag boolean，表示该给定的startKey是否是精确的开始扫描的值，如果不是，则是默认的开始扫描的位置
	 * @return List<NavigableMap<byte[], byte[]>>
	 */
	public List<NavigableMap<byte[], byte[]>> scanColumnKeyValueList(String tableName,String columnFamily,String startKey,String stopKey,int page,int pageSize,boolean rowFlag){
		ResultScanner resultScanner = null;
		logger.info("start rowKey:"+startKey);
		logger.info("stop stopKey:"+stopKey);
		try {
			Scan scan = new Scan();
			int cacheSize = 0;
			if(rowFlag){//如果精确开始扫描位置，则只返回limit行数据
				cacheSize = pageSize +1;
			}else{
				cacheSize = page*pageSize+1;
			}
//			scan.setBatch(cacheSize);
			scan.setCaching(cacheSize);
			 scan.setCacheBlocks(true);//RegionServer是否应当缓存 当前客户端访问过的数据块    如果是随机的get 这个最好为false
			 scan.setStartRow(Bytes.toBytes(startKey));
//			 scan.setStopRow(Bytes.toBytes(stopKey));
			 FilterList filterList = new FilterList();
			 Filter stopFilter =new InclusiveStopFilter(Bytes.toBytes(stopKey));
			 scan.addFamily(Bytes.toBytes(columnFamily));
//			 Filter valueFilter = new SingleColumnValueFilter(Bytes.toBytes(columnFamily), HBaseContent.topic, compareOp, value)//选定列簇和某一列，然后与列的value相比，正确的返回全部的row，注意如果某一行不含有该列，同样返回，除非通过filterIfColumnMissing 设置成真。
			 Filter pageFilter = new PageFilter(cacheSize);
			 filterList.addFilter(stopFilter);
			 filterList.addFilter(pageFilter);
		     scan.setFilter(filterList);
			 long a = System.currentTimeMillis();
			 HTable hTable = this.getHTable(tableName);
		     long b = System.currentTimeMillis();
		     resultScanner = hTable.getScanner(scan);
		     logger.info("query table timeMillis:"+(b-a)+"ms");
			 List<NavigableMap<byte[], byte[]>> keyValue = Collections.synchronizedList(new ArrayList<NavigableMap<byte[], byte[]>>());
			 for (Result result : resultScanner) {
//				 KeyValue[] raw = result.raw();
//				 for(int i=0;i<raw.length;i++){
//					 logger.info("rowKey:"+new String(raw[i].getKey()));
//				}
				 NavigableMap<byte[], byte[]> familyMap = result.getFamilyMap(Bytes.toBytes(HBaseContent.HBASE_FAMILY_TOPIC_CATEGORY));
				 keyValue.add(familyMap);
			 }
			 if(rowFlag){//如果精确开始扫描位置，
				 return keyValue;
			 }else{//
				 if(keyValue != null && !keyValue.isEmpty()){
					 List<NavigableMap<byte[], byte[]>> subList = keyValue.subList(page*pageSize-pageSize, keyValue.size());
					 return subList;
				 }else return null;
			 }
		} catch (IOException e) {
			e.printStackTrace();
			return null;
		}finally{
			 if(resultScanner!=null)	resultScanner.close();
		}
	}

	public void removeFamily(String tableName, String family) throws IOException {
		try {
			String tmp = this.fixColName(family);
			if (admin.isTableAvailable(tableName))
				admin.disableTable(tableName);
			this.admin.deleteColumn(tableName, tmp);
			this.admin.enableTable(tableName);
		} catch (IOException e) {
			throw e;
		}
	}


	/**
	 * 创建列的描述,添加后，该列会有一个冒号的后缀，用于存储(列)族, 将来如果需要扩展，那么就在该列后加入(列)族
	 * @param colName
	 * @return
	 */
	private HColumnDescriptor createHCDesc(String family) {
		String tmp = this.fixColName(family);
		byte[] colNameByte = Bytes.toBytes(tmp);
		return new HColumnDescriptor(colNameByte);
	}
	
	/**
	 * 针对hbase的列的特殊情况进行处理,列的情况: course: or course:math, 就是要么带列族，要么不带列族(以冒号结尾)
	 * 
	 * @param colName列
	 * @param cluster 列族
	 * @return
	 */
	private String fixColName(String colName, String cluster) {
		if (cluster != null && cluster.trim().length() > 0
				&& colName.endsWith(cluster)) {
			return colName;
		}
		String tmp = colName;
		int index = colName.indexOf(COLENDCHAR);
		// int leng = colName.length();
		if (index == -1) {
			tmp += COLENDCHAR;
		}
		// 直接加入列族
		if (cluster != null && cluster.trim().length() > 0) {
			tmp += cluster;
		}
		return tmp;
	}

	private String fixColName(String colName) {
		return this.fixColName(colName, null);
	}

	/**
	 * 创建表的描述
	 * 
	 * @param tableName
	 * @return
	 * @throws Exception
	 */
	private HTableDescriptor createHTDesc(final String tableName) {
		return new HTableDescriptor(tableName);
	}

	/**
	 * 取得某个表
	 * @param tableName
	 * @return
	 * @throws Exception
	 */
	private HTable getHTable(String tableName) throws IOException {
		try {
			return new HTable(configuration, tableName);
		} catch (IOException e) {
			throw e;
		}
	}

	/** 
     * 删除一张表 
     * @param tableName 
     */  
    public void dropTable(String tableName) {  
        try {  
            admin.disableTable(tableName);  
            admin.deleteTable(tableName);  
        } catch (MasterNotRunningException e) {  
            e.printStackTrace();  
        } catch (ZooKeeperConnectionException e) {  
            e.printStackTrace();  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
  
    }
    
    /** 
     * 创建表 
     * @param tableName 
     */  
    public void createTable(String tableName,String family) {  
        try {  
            if (admin.tableExists(tableName)) {
            	return ;
            }  else{
            	HTableDescriptor tableDescriptor = new HTableDescriptor(tableName);  
            	tableDescriptor.addFamily(new HColumnDescriptor(family));  
            	admin.createTable(tableDescriptor);  
            }
        } catch (MasterNotRunningException e) {  
            e.printStackTrace();  
        } catch (ZooKeeperConnectionException e) {  
            e.printStackTrace();  
        } catch (IOException e) {  
            e.printStackTrace();  
        }  
    }
    /**
     * 更新表中某个family的qualifier的value值
     * @param tableName 表名称
     * @param rowKey rowkey
     * @param columnFamily 列簇
     * @param qualifier 要更新的key值
     * @param value 要更新的value值
     * @throws IOException
     */
    public void updateQualifierValue(String tableName,String rowKey,String columnFamily,String qualifier,String value) throws IOException{
    	Map<String, String> column = Collections.synchronizedMap(new HashMap<String,String>());
    	column.put(qualifier, value);
    	this.insertAndUpdate(tableName, rowKey, columnFamily, column);
    }

}
